package com.atguigu.champter7;

import com.atguigu.beans.WaterSensor;
import com.atguigu.utils.AtguiguUtil;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;

public class Flink09_WaterMark_custom {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 9999);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        env
                .socketTextStream("hadoop104", 9999)
                .map(line -> {
                    String[] watersensors = line.split(",");
                    return new WaterSensor(watersensors[0], Long.valueOf(watersensors[1]), Integer.valueOf(watersensors[2]));
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forGenerator(new WatermarkStrategy<WaterSensor>() {
                                    @Override
                                    public WatermarkGenerator<WaterSensor> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                                        return new MyGenerator();
                                    }
                                })
                                .withTimestampAssigner((event, ts) -> event.getTs())
                )
                .keyBy(WaterSensor::getId)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(3)))
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String key
                            , Context ctx
                            , Iterable<WaterSensor> elements
                            , Collector<String> out) throws Exception {
                        List<WaterSensor> list = AtguiguUtil.toList(elements);
                        String msg = key + " " + list + " " + ctx.window();
                        out.collect(msg);
                    }
                })
                .print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class MyGenerator implements WatermarkGenerator {
        Long maxTimestamp = Long.MIN_VALUE + 1000;

        @Override
        //来一条数据执行一次
        public void onEvent(Object event, long eventTimestamp, WatermarkOutput output) {
            System.out.println("This is on Event");
            maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
            output.emitWatermark(new Watermark(maxTimestamp));
        }

        @Override
        //周期性执行[ 默认为200ms ],无论有无代码
        public void onPeriodicEmit(WatermarkOutput output) {
//            output.emitWatermark();
//            System.out.println("Here is onPeriodicEmit");
        }
    }
}
